package cn.t.modules.scheduler.core.schedule;

import cn.t.modules.scheduler.core.schedule.task.TaskFactory;
import cn.t.modules.scheduler.core.schedule.repository.po.SchedulePo;
import cn.t.modules.timewheel.core.hierarchical.TTaskScheduler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;

/**
 * 计划消费者(持有线程循环消费)
 * @author 陶敏麒
 * @date 2024/3/6 22:00
 */
@Component
@Slf4j
public class ScheduleConsumer {

    @Resource
    private ScheduleQueue scheduleQueue;
    @Resource
    private TTaskScheduler taskScheduler;

    private final List<Thread> consumerThreads = new ArrayList<>();

    public ScheduleConsumer(@Value("${schedule.consumer.threads:1}") int threads) {
        for (var i = 0; i < threads; i++) {
            this.consumerThreads.add(this.threadSupplier.get());
        }
    }

    Supplier<Thread> threadSupplier = () -> new Thread(() -> {
        log.info(">>>>___计划任务消费线程启动...");
        while (!Thread.currentThread().isInterrupted()) {
            try {
                var po = this.scheduleQueue.poll();
                if (po != null) {
                    this.consume(po);
                    TimeUnit.MICROSECONDS.sleep(1);
                }
            } catch (InterruptedException e) {
                log.warn("消费计划时被打断, {}", e.getMessage());
                // 重置打断标志位
                Thread.currentThread().interrupt();
            } catch (Exception e) {
                log.warn("消费计划时异常, {}", e.getMessage());
            }
        }
    }, "计划任务消费线程");

    /**
     * 启动消费者线程
     */
    @PostConstruct
    protected void raiseConsumer() {
        this.consumerThreads.forEach(Thread::start);
    }

    /**
     * 消费计划
     */
    private void consume(SchedulePo po) {
        var task = TaskFactory.buildTask(po);
        if (task == null) {
            return;
        }
        this.taskScheduler.addTask(task);
    }
}
